package com.shujia.flink.run;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;

public class FlinkSQLRunner {
    public static String loadSQLFile(String path) throws IOException {
        BufferedReader br = new BufferedReader(new FileReader(path));
        String sqlStr = "";
        String line;
        while ((line = br.readLine()) != null) {
            sqlStr = sqlStr + line + "\n";
        }
        return sqlStr;
    }

    public static void main(String[] args) throws IOException {
        /*
         * args接收两个参数：模式、SQL文件的路径
         */
        if (args.length != 2) {
            System.out.println("请指定两个参数：运行模式、路径！");
            return;
        }
        String runtimeMode = args[0];
        String sqlFilePath = args[1];

        System.out.println("=================================");
        System.out.println("正在处理的SQL文件：" + sqlFilePath);
        System.out.println("=================================");

        EnvironmentSettings envSettings;
        if ("stream".equals(runtimeMode)) {
            envSettings = EnvironmentSettings
                    .newInstance()
                    // 流处理模式
                    .inStreamingMode()
                    .build();
        } else if ("batch".equals(runtimeMode)) {
            envSettings = EnvironmentSettings
                    .newInstance()
                    // 批处理模式
                    .inBatchMode()
                    .build();
        } else {
            System.out.println("模式指定出错：" + runtimeMode);
            return;
        }


        TableEnvironment tableEnv = TableEnvironment.create(envSettings);

        // 加载SQL文件
        String sqlStr = loadSQLFile(sqlFilePath);

        // 依次执行每一条SQL
        for (String sql : sqlStr.split(";")) {
            if (!sql.trim().isEmpty()) {
                System.out.println("正在执行的SQL为：" + sql);
                tableEnv.executeSql(sql);
            }
        }

    }
}
